package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

public class Demo2Api {

    Connection connection;

    /**
     * 创建链接
     */
    @Before
    public void init() {
        //1、创建hbase链接
        //创建hadoop配置对象
        Configuration configuration = HBaseConfiguration.create();
        //指定链接参数
        //指定zk地址，在zk中保存了hbase元数据
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        //建立链接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 创建表
     */
    @Test
    public void createTable() throws Exception {
        //获取Admin,用于创建修改删除表
        Admin admin = connection.getAdmin();
        //创建表描述对象
        HTableDescriptor hTableDescriptor = new HTableDescriptor("student");
        //创建列簇描述对象
        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor("info");
        //增加列簇
        hTableDescriptor.addFamily(hColumnDescriptor);
        //创建表
        admin.createTable(hTableDescriptor);
    }

    /**
     * 删除表
     */
    @Test
    public void deleteTable() throws Exception {
        Admin admin = connection.getAdmin();
        //警用表
        admin.disableTable(TableName.valueOf("student"));
        //删除表
        admin.deleteTable(TableName.valueOf("student"));
    }

    /**
     * 插入单条数据
     */

    @Test
    public void put() throws Exception {
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student"));
        //创建put
        Put put = new Put(Bytes.toBytes("001"));
        //设置列名和列值
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("zhangsan"));
        //插入数据
        student.put(put);
    }

    /**
     * 通过rowkey查询数据
     */
    @Test
    public void get() throws Exception {
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student"));
        //构建get对象
        Get get = new Get(Bytes.toBytes("001"));
        //查询数据
        Result result = student.get(get);
        //解析数据
        byte[] value = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"));
        String name = Bytes.toString(value);
        System.out.println(name);
    }

    /**
     * 批量插入数据
     */
    @Test
    public void putStudents() throws Exception {
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student_split"));

        //构建一个list集合
        ArrayList<Put> puts = new ArrayList<>();

        //1、使用io流读取本地文件
        BufferedReader bufferedReader = new BufferedReader(new FileReader("../data/students.txt"));
        String line;
        while ((line = bufferedReader.readLine()) != null) {
            //取出数据
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];

            //构建put对象，以学号作为rowkey, 其他字段作为列
            Put put = new Put(Bytes.toBytes(id));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(name));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(age));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes(gender));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("clazz"), Bytes.toBytes(clazz));

            //将put对象保存到list集合中
            puts.add(put);
        }
        bufferedReader.close();

        //批量插入数据
        student.put(puts);
    }


    /**
     * scan全表扫描
     */
    @Test
    public void scan() throws Exception {
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student"));
        Scan scan = new Scan();
        //全表扫描
        ResultScanner scanner = student.getScanner(scan);

        //获取所有数据，得到一个迭代器
        Iterator<Result> iterator = scanner.iterator();
        while (iterator.hasNext()) {
            //去除一行
            Result result = iterator.next();
            //去除rowkey
            String id = Bytes.toString(result.getRow());
            //解析数据
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }
    }

    @Test
    public void delete()throws Exception{
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student"));
        Delete delete = new Delete(Bytes.toBytes("1500101000"));
        //删除一行
        student.delete(delete);
    }

}
